MultiCast.py

Set Multicast Mode

It shows how to set transmission mode to multicast mode, that is, querying the user to launch the multicast controlling application or the multicast monitoring application, and opening camera and setting multicast mode in the specified access permission by MV_GIGE_SetTransmissionType().

# -- coding: utf-8 --

import sys
import threading
import platform
import os
from ctypes import *

# Compatible with different operating systems to load DDL
currentsystem = platform.system()
if currentsystem == 'Windows':
    sys.path.append(os.path.join(os.getenv('MVCAM_COMMON_RUNENV'), "Samples", "Python", "MvImport"))
else:
    sys.path.append(os.path.join("..", "..", "MvImport"))
from MvCameraControl_class import *

# Compatible with input processing of Python 2.X and 3.X
if sys.version_info[0] < 3: 
    # Python 2.x
    input_func = raw_input
else: 
    # Python 3.x
    input_func = input

# Decoding Characters
def decoding_char(ctypes_char_array):
    """
    Safely decode a string from a ctypes character array.
    Compatible with Python 2.x and 3.x, as well as 32-bit and 64-bit environments.
    """
    byte_str = memoryview(ctypes_char_array).tobytes()
    
    # Truncate at the first null character
    null_index = byte_str.find(b'\x00')
    if null_index != -1:
        byte_str = byte_str[:null_index]
    
    # Attempt to decode using multiple encodings
    for encoding in ['gbk', 'utf-8', 'latin-1']:
        try:
            return byte_str.decode(encoding)
        except UnicodeDecodeError:
            continue
    
    # If all encodings fail, use a replacement strategy
    return byte_str.decode('latin-1', errors='replace')



g_bExit = False

# Customize the thread function
def work_thread(cam=0, pData=0, nDataSize=0):
    stOutFrame = MV_FRAME_OUT()  
    memset(byref(stOutFrame), 0, sizeof(stOutFrame))
    while True:
        ret = cam.MV_CC_GetImageBuffer(stOutFrame, 1000)
        if None != stOutFrame.pBufAddr and 0 == ret:
            print ("get one frame: Width[%d], Height[%d], nFrameNum[%d]"  % (stOutFrame.stFrameInfo.nWidth, stOutFrame.stFrameInfo.nHeight, stOutFrame.stFrameInfo.nFrameNum))
            cam.MV_CC_FreeImageBuffer(stOutFrame)
        else:
            print ("no data[0x%x]" % ret)
        if g_bExit == True:
            break

if __name__ == "__main__":
    
    try:
        # Initialize SDK resources
        MvCamera.MV_CC_Initialize()

        SDKVersion = MvCamera.MV_CC_GetSDKVersion()
        print ("SDKVersion[0x%x]" % SDKVersion)
            
        deviceList = MV_CC_DEVICE_INFO_LIST()
        tlayerType = MV_GIGE_DEVICE 
        
        # Enumerate devices
        ret = MvCamera.MV_CC_EnumDevices(tlayerType, deviceList)
        if ret != 0:
            print ("enum devices fail! ret[0x%x]" % ret)
            sys.exit()

        if deviceList.nDeviceNum == 0:
            print ("find no device!")
            sys.exit()

        print ("find %d devices!" % deviceList.nDeviceNum)

        for i in range(0, deviceList.nDeviceNum):
            mvcc_dev_info = cast(deviceList.pDeviceInfo[i], POINTER(MV_CC_DEVICE_INFO)).contents
            if mvcc_dev_info.nTLayerType == MV_GIGE_DEVICE:
                print ("\ngige device: [%d]" % i)
                strModeName = decoding_char(mvcc_dev_info.SpecialInfo.stGigEInfo.chModelName)
                print ("device model name: %s" % strModeName)
                strSerialNumber = decoding_char(mvcc_dev_info.SpecialInfo.stGigEInfo.chSerialNumber)
                print("device serial number: %s" % strSerialNumber)
                nip1 = ((mvcc_dev_info.SpecialInfo.stGigEInfo.nCurrentIp & 0xff000000) >> 24)
                nip2 = ((mvcc_dev_info.SpecialInfo.stGigEInfo.nCurrentIp & 0x00ff0000) >> 16)
                nip3 = ((mvcc_dev_info.SpecialInfo.stGigEInfo.nCurrentIp & 0x0000ff00) >> 8)
                nip4 = (mvcc_dev_info.SpecialInfo.stGigEInfo.nCurrentIp & 0x000000ff)
                print ("current ip: %d.%d.%d.%d\n" % (nip1, nip2, nip3, nip4))
                

        nConnectionNum = input_func("please input the number of the device to connect:")

        if int(nConnectionNum) >= deviceList.nDeviceNum:
            print ("intput error!")
            sys.exit()

        # Create the camera instance
        cam = MvCamera()

        # Select a device, and create a handle
        stDeviceList = cast(deviceList.pDeviceInfo[int(nConnectionNum)], POINTER(MV_CC_DEVICE_INFO)).contents

        ret = cam.MV_CC_CreateHandle(stDeviceList)
        if ret != 0:
            raise Exception ("create handle fail! ret[0x%x]" % ret)

        # Ask the user to launch the multicast controlling application or the multicast monitoring application
        #en:Ask the user to launch: the multicast controlling application or the multicast monitoring application.
        print ("start multicast sample in (c)ontrol or in (m)onitor mode? (c/m)")

        key = input_func()

        # Search for the mode used by the user
        monitor = False
        if key == 'm' or key == 'M':
            monitor = True
        elif key == 'c' or key == 'C':
            monitor = False
        else:
            raise Exception ("intput error!")

        if monitor:
            ret = cam.MV_CC_OpenDevice(MV_ACCESS_Monitor, 0)
            if ret != 0:
                raise Exception ("open device fail! ret[0x%x]" % ret)
        else:
            ret = cam.MV_CC_OpenDevice(MV_ACCESS_Control, 0)
            if ret != 0:
                raise Exception ("open device fail! ret[0x%x]" % ret)

        # Get optimal packet size (only supported by GigE devices)
        if stDeviceList.nTLayerType == MV_GIGE_DEVICE:
            nPacketSize = cam.MV_CC_GetOptimalPacketSize()
            if int(nPacketSize) > 0:
                ret = cam.MV_CC_SetIntValue("GevSCPSPacketSize",nPacketSize)
                if ret != 0:
                    print ("Warning: Set Packet Size fail! ret[0x%x]" % ret)
            else:
                print ("Warning: Get Packet Size fail! ret[0x%x]" % nPacketSize)

        # Specify the multicast IP address
        strIp = "239.0.1.23"
        device_ip_list = strIp.split('.')
        dest_ip = (int(device_ip_list[0]) << 24) | (int(device_ip_list[1]) << 16) | (int(device_ip_list[2]) << 8) | int(device_ip_list[3])
        print ("dest ip: %s" % strIp)

        # Specify the multicast port
        stTransmissionType = MV_TRANSMISSION_TYPE() 
        memset(byref(stTransmissionType), 0, sizeof(MV_TRANSMISSION_TYPE))

        stTransmissionType.enTransmissionType = MV_GIGE_TRANSTYPE_MULTICAST
        stTransmissionType.nDestIp = dest_ip
        stTransmissionType.nDestPort = 8787

        ret = cam.MV_GIGE_SetTransmissionType(stTransmissionType)
        if MV_OK != ret:
            raise Exception ("set transmission type fail! ret [0x%x]" % ret)

        # Start grabbing images
        ret = cam.MV_CC_StartGrabbing()
        if ret != 0:
            raise Exception ("start grabbing fail! ret[0x%x]" % ret)

        try:
            hThreadHandle = threading.Thread(target=work_thread, args=(cam, None, None))
            hThreadHandle.start()
        except:
            raise Exception ("error: unable to start thread")
            
        print ("press Enter key to stop grabbing.")
        input_func()

        g_bExit = True
        hThreadHandle.join()

        # Stop grabbing images
        ret = cam.MV_CC_StopGrabbing()
        if ret != 0:
            raise Exception ("stop grabbing fail! ret[0x%x]" % ret)

        # Turn off the device
        ret = cam.MV_CC_CloseDevice()
        if ret != 0:
            raise Exception ("close deivce fail! ret[0x%x]" % ret)

        # Destroy the handle
        cam.MV_CC_DestroyHandle()
    except Exception as e:
        print(e)
        cam.MV_CC_CloseDevice()
        cam.MV_CC_DestroyHandle()
    finally:
        # Release SDK resources
        MvCamera.MV_CC_Finalize()